/*
package com.xunyin.nettyV;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;


public class BootMqttMsgBack {

    private static Logger log = LoggerFactory.getLogger(BootMqttMsgBack.class);
    public static Map<Channel, Set<String>> channelMqttMessageMap = new HashMap<>();

    */
/**
 * 确认连接请求
 *
 * @param channel
 * @param mqttMessage
 * <p>
 * 根据qos发布确认
 * @param channel
 * @param mqttMessage
 * <p>
 * 发布完成 qos2
 * @param channel
 * @param mqttMessage
 * <p>
 * 订阅确认
 * @param channel
 * @param mqttMessage
 * <p>
 * 取消订阅确认
 * @param channel
 * @param mqttMessage
 * <p>
 * 心跳响应
 * @param channel
 * @param mqttMessage
 *//*

    public static void connack(Channel channel, MqttMessage mqttMessage) {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();

        //	构建返回报文， 可变报头
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
        log.info("back--" + connAck.toString());
        channel.writeAndFlush(connAck);
    }

    */
/**
 * 根据qos发布确认
 *
 * @param channel
 * @param mqttMessage
 *//*

    public static void puback(Channel channel, MqttMessage mqttMessage) throws MqttException {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        System.out.println("publish data--" + data);
        String payloadStr = data;
        JSONObject jsonPayloadObject = JSON.parseObject(payloadStr);
        // 硬件物理地址
        String mac = jsonPayloadObject.getString("mac");
        System.out.println("硬件-mac:"+mac);
        byte[] bytes = data.getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = Unpooled.wrappedBuffer(bytes);

        switch (qos) {
            case AT_MOST_ONCE:        // 至多一次
                System.out.println("AT_MOST_ONCE");
                //	构建返回报文， 可变报头
                MqttPublishVariableHeader mqttMessageIdVariableHeaderBack0 =new MqttPublishVariableHeader(mqttPublishMessage.variableHeader().topicName(),mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack0 = new MqttFixedHeader(MqttMessageType.PUBLISH, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建publish消息体
                MqttPublishMessage publishMessage0 = new MqttPublishMessage(mqttFixedHeaderBack0, mqttMessageIdVariableHeaderBack0,
                        buf);
                String topicName =mqttPublishMessage.variableHeader().topicName();
                log.info("接收到topicName："+topicName);
                if(topicName.contains("/basic/up")){
                    //SubMsg.subMsg("client-id-0", topicName);
                    JSONObject reBasicDown = new JSONObject();
                    reBasicDown.put("mac",mac);
                    //reBasicDown.put("relay",0);
                    reBasicDown.put("time", DateUtils.parseDate2String(new Date(), "yyyy-MM-dd HH:mm:ss"));
                    reBasicDown.put("reboot",1);
                    reBasicDown.put("ip", MqConfig.ip);
                    reBasicDown.put("port", MqConfig.port);
                    // 给硬件发送消息-前提是硬件订阅这主题
                    //PubMsg.publish(reBasicDown.toJSONString(), "client-id-"+mac, "/dev/"+mac+"/basic/down");
                }
                log.info("publish--" + publishMessage0.toString());
                publish(publishMessage0,mqttPublishMessage);


                break;
            case AT_LEAST_ONCE:        // 至少一次
                System.out.println("AT_LEAST_ONCE");
                //	构建返回报文， 可变报头
                MqttPublishVariableHeader mqttMessageIdVariableHeaderBack =new MqttPublishVariableHeader(mqttPublishMessage.variableHeader().topicName(),mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBLISH, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_LEAST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建publish消息体
                MqttPublishMessage publishMessage = new MqttPublishMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack,
                        buf);
                String topicName1 =mqttPublishMessage.variableHeader().topicName();
                log.info("接收到topicName1："+topicName1);
                if(topicName1.contains("/basic/up")){
                     //SubMsg.subMsg("client-id-"+mac, topicName1);
                     log.info("硬件物理地址："+mac);
                    //JSONObject reBasicDown = new JSONObject();
                    //reBasicDown.put("mac",mac);
                    //reBasicDown.put("relay",0);
                    //reBasicDown.put("time", DateUtils.parseDate2String(new Date(), "yyyy-MM-dd HH:mm:ss"));
                    //reBasicDown.put("reboot",0);
                    //reBasicDown.put("ip", MqConfig.ip);
                    //reBasicDown.put("port", MqConfig.port);
                    // 给硬件发送消息-前提是硬件订阅这主题
                    //PubMsg.publish(reBasicDown.toJSONString(), "client-id-0", "/dev/"+mac+"/basic/down");
                }
                log.info("publish--" + publishMessage.toString());
                publish(publishMessage,mqttPublishMessage);
                break;
            case EXACTLY_ONCE:        //	刚好一次
                //	构建返回报文， 固定报头
//                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
//                //	构建返回报文， 可变报头
//                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
//                MqttPubAckMessage mqttMessageBack = new MqttPubAckMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
//                log.info("back--" + mqttMessageBack.toString());

//                arrayList.get(0).writeAndFlush(mqttMessageBack);
//
//                for (Channel channel1 : arrayList) {
//                    //if (channel1 == channel){
//                        channel1.writeAndFlush(mqttMessageBack);
//                    //}
//                }

//                channel.writeAndFlush(mqttMessageBack);


                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack21 = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.EXACTLY_ONCE, false, 0x02);
                //	构建返回报文， 可变报头

                //MqttPublishVariableHeader mqttMessageIdVariableHeaderBack21 = new MqttPublishVariableHeader(mqttPublishMessage.variableHeader().packetId());
                MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(mqttPublishMessage.variableHeader().topicName(), mqttPublishMessage.variableHeader().packetId());
                MqttPublishMessage mqttMessageBack2 = new MqttPublishMessage(mqttFixedHeaderBack21,
                        mqttPublishVariableHeader,
                        buf);
                log.info("MqttPublishMessage--" + mqttMessageBack2.toString());
                publish(mqttMessageBack2,mqttPublishMessage);

                //channel.writeAndFlush(mqttMessageBack2);
                break;
            default:
                break;
        }
    }

    public static void publish(MqttPublishMessage mqttMessagePublish,MqttPublishMessage mqttPublishMessage){
        ReferenceCountUtil.retain(mqttMessagePublish);
        for (Channel channel1 : channelMqttMessageMap.keySet()) {
            Set<String> strings = channelMqttMessageMap.get(channel1);
            for (String string : strings) {
                String topicName = mqttPublishMessage.variableHeader().topicName();
                if (string.equals(topicName)){
                    channel1.writeAndFlush(mqttMessagePublish);
                }
            }
        }
    }

    */
/**
 * 发布完成 qos2
 *
 * @param channel
 * @param mqttMessage
 *//*

    public static void pubcomp(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        log.info("back--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    */
/**
 * 订阅确认
 *
 * @param channel
 * @param mqttMessage
 *//*

    public static void suback(Channel channel, MqttMessage mqttMessage) {

        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
        //log.info(topics.toString());
        List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        //	构建返回报文	有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 2 + topics.size());
        //	构建返回报文	订阅确认
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        log.info("back--" + subAck.toString());
        //channelMqttMessageMap.put(channel,topics);
        boolean b = channelMqttMessageMap.containsKey(channel);
        if (b) {
            for (String topic : topics) {
                channelMqttMessageMap.get(channel).add(topic);
            }
        } else {
            channelMqttMessageMap.put(channel, topics);
        }
        channel.writeAndFlush(subAck);
    }

    */
/**
 * 取消订阅确认
 *
 * @param channel
 * @param mqttMessage
 *//*

    public static void unsuback(Channel channel, MqttMessage mqttMessage) {
//        List<String> stringList = (List<String>) mqttMessage.payload();
//        for (String s : stringList) {
//            System.out.println(s);
//        }
        Object payload = mqttMessage.payload();
        String string = payload.toString();
        String topic = StringUtils.substringAfterLast(string, "=");

        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
        log.info("back--" + unSubAck.toString());

        for (Channel channel1 : channelMqttMessageMap.keySet()) {
            boolean contains = channelMqttMessageMap.get(channel1).contains(topic);
            if (contains) {
                channelMqttMessageMap.remove(channel1);
            }
        }

        channel.writeAndFlush(unSubAck);
    }

    */
/**
 * 心跳响应
 *
 * @param channel
 * @param mqttMessage
 *//*

    public static void pingresp(Channel channel, MqttMessage mqttMessage) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        log.info("back--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }


    public static void disconnect(Channel channel, MqttMessage mqttMessage) {
        channelMqttMessageMap.remove(channel);
    }
}*/
